package events

import (
	"Demo/book/interactive/repository"
	logger2 "Demo/book/pkg/logger"
	"Demo/book/pkg/saramax"
	"context"
	"github.com/IBM/sarama"
	"time"
)

type InteractiveReadEventBatchConsumer struct {
	l      logger2.LoggerV1
	client sarama.Client
	repo   repository.InteractiveRepository
}

func (i *InteractiveReadEventBatchConsumer) Start() error {
	client, err := sarama.NewConsumerGroupFromClient("interactive", i.client)
	if err != nil {
		panic(err)
	}
	go func() {
		errs := client.Consume(context.Background(), []string{"article_read"}, saramax.NewBatchConsumer[ReadEvent](i.l, i.Consume))
		if errs != nil {
			i.l.Error("推出了消费循环一场", logger2.Error(err))
		}
	}()
	return err
}

// 这个不是幂等的
func (i *InteractiveReadEventBatchConsumer) Consume(msgs []*sarama.ConsumerMessage, ts []ReadEvent) error {
	//在这里处理消息
	ids := make([]int64, 0, len(ts))
	bizs := make([]string, 0, len(ts))
	for _, evt := range ts {
		ids = append(ids, evt.Aid)
		bizs = append(bizs, "article")
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()
	//
	err := i.repo.BatchIncrReadCnt(ctx, bizs, ids)
	if err != nil {
		i.l.Error("批量增加阅读数失败", logger2.Error(err),
			logger2.Field{Name: "ids", Value: ids})
	}
	return nil
}
